package algebra.cost;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
//import java.util.TreeMap;

import reasoner.KBReason;

import utils.exception.CoreReasoningException;
import utils.exception.CostEstimateException;
import utils.exception.SyntaxParsingException;

import entities.concept.Concept;
import kb.DescriptionLogicsKB;
import algebra.op.*;
import algebra.query.ObjectQuery;
import algebra.query.indexing.SecondaryIndex;
import algebra.query.pd.PdConcept;
import algebra.query.pd.PdConjunction;
import algebra.query.pd.ProjDesc;
import algebra.rewriting.RewriteRule;
import algebra.rewriting.RuleService;
import algebra.rewriting.kfree.KFreeApproximate;
import algebra.rewriting.kfree.KFreeRewriting;
import algebra.rewriting.kfree.LangPd;
import algebra.rewriting.relational.CacheIntro;

public class PlanEnumeration {

	private final DescriptionLogicsKB kb;
	private final Concept cond;
	private final ProjDesc pd;
	private final PlanEstimation est;
	private final Set<SecondaryIndex> sisForUse;
	
	public static final boolean useMoreGeneralSelectionRemovingCondition = true;

	// private final static double scanCostInRemovedSelection = 2.0;
	// private final static double projectionCostInNestedSearch = 2.0;

	// by default, we use all SIs in the cache
	public PlanEnumeration(ObjectQuery query, DescriptionLogicsKB kb) {
		this(query, SecondaryIndex.allCachedQueries.keySet(), kb);
	}

	public PlanEnumeration(ObjectQuery query, Set<SecondaryIndex> givenSIs,
			DescriptionLogicsKB kb) {
		this.kb = kb;
		this.cond = query.getC();
		this.pd = query.getPd();
		this.sisForUse = new HashSet<SecondaryIndex>();
		this.sisForUse.addAll(givenSIs);
//		System.err.println("Total # of cached queries: "+this.sisForUse.size());
		this.est = new PlanEstimation(kb);
	}

	public Concept getQC() {
		return this.cond;
	}

	public ProjDesc getPd() {
		return this.pd;
	}

	public DescriptionLogicsKB getKB() {
		return this.kb;
	}

	public Set<String> getRestrictedConstants(AlgebraicOP query) {
		Set<String> ret = new HashSet<String>();
		ret.addAll(query.extractConstants());
		for (SecondaryIndex si : SecondaryIndex.allCachedQueries.keySet()) {
			ret.addAll(si.getPlanExpression().extractConstants());
		}
		return ret;
	}

	// return a single plan selected by the system
	public Object[] systemSelectedPlan() {
		Object[] res = new Object[2];
		Map<AlgebraicOP, Double> plans = searchPlan();
		Double min = Double.MAX_VALUE;
		AlgebraicOP leastplan = null;
		for (AlgebraicOP op : plans.keySet()) {
			Double cost = plans.get(op);
			if (cost < min) {
				min = cost;
				leastplan = op;
			}
		}
		res[0] = leastplan;
		res[1] = min;
		return res;
	}
	
	private void clearCache(){
		KFreeRewriting.clearCacheConstantsAndResetCounter();
		CacheIntro.clearCachedCp();
		LangPd.resetCache();
		CacheIntro.clearCachedCp();
	}

	public Map<AlgebraicOP, Double> searchPlan() {
		clearCache();
		Map<AlgebraicOP, Double> results = new HashMap<AlgebraicOP, Double>();
		AlgebraicOP rootplan = new ProjectionOP(pd, new SelectionOP(
				new PrimaryOP(kb), cond, kb), kb);
		// find and set all relevant constants, i.e., those in Q and Si
		KFreeRewriting.setConstants(this.getRestrictedConstants(rootplan));
//		System.out.println("\nCaptured constants: "
//				+ KFreeRewriting.getCapturedConstants().size());
		//
		Map<AlgebraicOP, Map<AlgebraicOP, Double>> plans = new HashMap<AlgebraicOP, Map<AlgebraicOP, Double>>();
		Map<AlgebraicOP, AlgebraicOP> bestplans = new HashMap<AlgebraicOP, AlgebraicOP>();
		try {
			this.computePlans(rootplan, plans, bestplans);
			for (AlgebraicOP op : plans.get(rootplan).keySet()) {
				if (!results.containsKey(op)) {
					results.put(op, plans.get(rootplan).get(op));
				}
			}
		} catch (CostEstimateException e) {
			e.printStackTrace();
		}
		return results;
	}
	
	private boolean usefulForKFreeSelectionOrProjection(Set<ScanOP> scans) 
			throws SyntaxParsingException, CoreReasoningException{
		ProjDesc pdc = new PdConcept(this.cond);
		ProjDesc pdscan = null;
		for(ScanOP op:scans){
			ProjDesc sub = op.getSecondaryIndex().getPdi();
			if(pdscan == null){
				pdscan = sub;
			}else{
				pdscan = new PdConjunction(pdscan, sub);
			}
		}
		boolean ok = KFreeApproximate.isIncluded(pdc, pdscan);
		if(ok){
			return true;
		}
		ok = KFreeApproximate.isIncluded(this.pd, pdscan);
		return ok;
	}

	private boolean usefulForSelectionRemove(Set<ScanOP> scans) {
		// get all Scans' subqueries
		Set<Concept> cs = new HashSet<Concept>();
		for (ScanOP si : scans) {
			if (si.getSubquery() instanceof ConstantOP) {
				cs.add(si.getSubquery().asConstantOP().getQueryCocnept());
				//optimization: consider Si's concept as well. July 2013.
				if(useMoreGeneralSelectionRemovingCondition){
					cs.add(si.getSecondaryIndex().getCi());
				}
				//
			} else {
				return false;
			}
		}
		try {
			return new KBReason().subsumesGuarded(
					Concept.getConjunctiveConcept(cs), this.cond, kb);
		} catch (CoreReasoningException e) {
			e.printStackTrace();
		}
		return false;
	}

//	private boolean usefulForProjectionRemove(Set<ScanOP> scans) {
//		// /each Scan has a more general pd than the query pd
//		for (ScanOP si : scans) {
//			ProjDesc pdsi = si.getSecondaryIndex().getPdi();
//			if (!LangPd.moreGeneral(pdsi, pd)) {
//				return false;
//			}
//		}
//		return true;
//	}

	// System-R style optimization: not only the best plan, but potential plans
	// that can remove selection
	private boolean potentiallyUsefulScans(AlgebraicOP op, AlgebraicOP cand) {
		boolean ok = false;
		if (!(op instanceof PrimaryOP)) {
			return false;
		}
		if (!cand.hasOnlyScanOP()) {
			return false;
		}
		Set<ScanOP> scans = cand.getAllScanOP();
		ok = usefulForSelectionRemove(scans);
		if(ok){
			return true;
		}
		try {
			ok = usefulForKFreeSelectionOrProjection(scans);
		} catch (Exception e) {
			e.printStackTrace();
		}

		return ok;
	}

	private void addToPlanCost(AlgebraicOP op,
			Map<AlgebraicOP, Map<AlgebraicOP, Double>> map, AlgebraicOP plan,
			Double cost) {
		Map<AlgebraicOP, Double> ret = null;
		if (map.containsKey(op)) {
			ret = map.get(op);
		} else {
			ret = new HashMap<AlgebraicOP, Double>();
		}
		if (!ret.containsKey(plan)) {
			// do not update cost
			ret.put(plan, cost);
		}
		map.put(op, ret);
	}

	private void bestForPI(PrimaryOP pi,
			Map<AlgebraicOP, Map<AlgebraicOP, Double>> plans,
			Map<AlgebraicOP, AlgebraicOP> bestplans)
			throws CostEstimateException {
		Set<AlgebraicOP> res = this.applyRewritingSingle(pi);
		AlgebraicOP bestcand = null;
		double mincost = Double.MAX_VALUE;
		for (AlgebraicOP alt : res) {
			if (alt.equals(pi)) {
				double curcost = est.getPlanCost(alt);
				if (curcost < mincost) {
					mincost = curcost;
					bestcand = alt;
				}
			} else {
				if (!plans.containsKey(alt)) {
					computePlans(alt, plans, bestplans);
				}
				for (AlgebraicOP cand : plans.get(alt).keySet()) {
					double curcost = 0;
					curcost = plans.get(alt).get(cand);
					if (potentiallyUsefulScans(pi, cand)) {
						// add promising (though higher cost) scans as well
						addToPlanCost(pi, plans, cand, curcost);
						if (curcost < mincost) {
							mincost = curcost;
							bestcand = cand;
						}
					} else if (curcost < mincost) {
						mincost = curcost;
						bestcand = cand;
					}
				}
			}

		}
		addToPlanCost(pi, plans, bestcand, mincost);
		bestplans.put(pi, bestcand);
	}

	private void bestForProjection(ProjectionOP pj,
			Map<AlgebraicOP, Map<AlgebraicOP, Double>> plans,
			Map<AlgebraicOP, AlgebraicOP> bestplans)
			throws CostEstimateException {
		double mincost = Double.MAX_VALUE;
		AlgebraicOP bestcand = null;
		AlgebraicOP sub = pj.getSubquery();
		computePlans(sub, plans, bestplans);
		for (AlgebraicOP psub : plans.get(sub).keySet()) {
			double subcost = plans.get(sub).get(psub);
			ProjectionOP newpj = new ProjectionOP(pj.getPrjDesc(), psub,
					pj.getPlanEvalKB());
			// get all possible new plans for this new projection
			Set<AlgebraicOP> in = new HashSet<AlgebraicOP>();
			in.add(newpj);
			Set<AlgebraicOP> res = this.filterResults(newpj,
					applyRulesSaturate(in));
			// Set<AlgebraicOP> res = applyRewritingSingle(newpj);
			for (AlgebraicOP pjcand : res) {
				double cost = 0;
				if ((pjcand instanceof ScanOP)) {
					// the projection can be removed
					// System.err.println("\n\nRemoved projection for "+newpj);
					cost = subcost;
					addToPlanCost(pj, plans, pjcand, cost);
				} else if (pjcand instanceof ProjectionOP
						&& newpj.getSubquery() instanceof JoinOP
						&& pjcand.asProjectionOP().getSubquery() instanceof ScanOP) {
					// a result of nested search, always keep it
					ScanOP sp = pjcand.asProjectionOP().getSubquery()
							.asScanOP();
					if (!plans.containsKey(sp)) {
						computePlans(sp, plans, bestplans);
					}
					// System.out.println("best for "+sp+"\n "+bestplans.get(sp));
					for (AlgebraicOP newsp : plans.get(sp).keySet()) {
						ProjectionOP newpjcand = new ProjectionOP(pjcand
								.asProjectionOP().getPrjDesc(), newsp, pjcand
								.asProjectionOP().getPlanEvalKB());
						cost = plans.get(sp).get(newsp)
								+ est.getPlanCost(newpjcand);
//						cost = est.getPlanCost(newpjcand);
						addToPlanCost(pj, plans, newpjcand, cost);
					}
					// cost = projectionCostInNestedSearch;
					// addToPlanCost(pj, plans, pjcand, cost);
				} else {
					cost = subcost + est.getPlanCost(pjcand);
//					cost = est.getPlanCost(pjcand);
					addToPlanCost(pj, plans, pjcand, cost);
				}
				if (cost < mincost) {
					mincost = cost;
					bestcand = pjcand;
				}
			}
			// for a particular pjcand, only use its least cost plan as the best
			// plan
			// i.e. local best
			addToPlanCost(pj, plans, bestcand, mincost);
		}
		// i.e. global best
		bestplans.put(pj, bestcand);
	}

	private void bestForSelection(SelectionOP sp,
			Map<AlgebraicOP, Map<AlgebraicOP, Double>> plans,
			Map<AlgebraicOP, AlgebraicOP> bestplans)
			throws CostEstimateException {
		AlgebraicOP sub = sp.getSubquery();
		if (!plans.containsKey(sub)) {
			computePlans(sub, plans, bestplans);
		}
		double mincost = Double.MAX_VALUE;
		AlgebraicOP bestcand = null;
		for (AlgebraicOP psub : plans.get(sub).keySet()) {
			double subcost = plans.get(sub).get(psub);
			if(subcost > mincost){
				//if the subcost is already larger than the min cost, don't ever try this plan
				// since the final plan is at least as costly as the subplan;
				continue;
			}
			SelectionOP newsp = new SelectionOP(psub,
					sp.getSelectionCondition(), sp.getPlanEvalKB());
			// get all possible new plans for this new projection
			Set<AlgebraicOP> in = new HashSet<AlgebraicOP>();
			in.add(newsp);
			Set<AlgebraicOP> res = this.filterResults(newsp,
					applyRulesSaturate(in));
			// Set<AlgebraicOP> res = applyRewritingSingle(newsp);
			for (AlgebraicOP spcand : res) {
				double cost = 0;
				if (!(spcand instanceof SelectionOP)) {
					// selection removed! cost to be adjusted.
//					cost = subcost;
					cost = est.getPlanCost(spcand);
					// cost = scanCostInRemovedSelection;
					addToPlanCost(sp, plans, spcand, cost);
				} else {
					cost = est.getPlanCost(spcand.asSelectionOP().getSubquery()) + est.getPlanCost(spcand);
//					cost = est.getPlanCost(spcand);
					addToPlanCost(sp, plans, spcand, cost);
				}
				if (cost < mincost) {
					mincost = cost;
					bestcand = spcand;
				}
			}
			// for a particular pjcand, only use its least cost plan as the best
			// plan
			// i.e. local best
			addToPlanCost(sp, plans, bestcand, mincost);
		}
		// i.e. global best
		bestplans.put(sp, bestcand);
	}

	private void bestForScan(ScanOP sp,
			Map<AlgebraicOP, Map<AlgebraicOP, Double>> plans,
			Map<AlgebraicOP, AlgebraicOP> bestplans)
			throws CostEstimateException {
		AlgebraicOP sub = sp.getSubquery();
		if (!plans.containsKey(sub)) {
			computePlans(sub, plans, bestplans);
		}
		double mincost = Double.MAX_VALUE;
		AlgebraicOP bestcand = null;
		for (AlgebraicOP psub : plans.get(sub).keySet()) {
			double subcost = plans.get(sub).get(psub);
			ScanOP newsp = new ScanOP(psub, sp.getSecondaryIndex());
			// get all possible new plans for this new projection
			Set<AlgebraicOP> in = new HashSet<AlgebraicOP>();
			in.add(newsp);
			Set<AlgebraicOP> res = this.filterResults(newsp,
					applyRulesSaturate(in));
			// Set<AlgebraicOP> res = applyRulesSaturate(in);
			for (AlgebraicOP pjcand : res) {
				double cost = 0;
				if (pjcand instanceof SelectionOP) {
					// scan may become selection
					SelectionOP sel = pjcand.asSelectionOP();
					AlgebraicOP selsub = sel.getSubquery();
					if (!plans.containsKey(selsub)) {
						computePlans(selsub, plans, bestplans);
					}
					for (AlgebraicOP allselsub : plans.get(selsub).keySet()) {
						cost = est.getPlanCost(sel)
								+ plans.get(selsub).get(allselsub);
						SelectionOP newsel = new SelectionOP(allselsub,
								sel.getSelectionCondition(),
								sel.getPlanEvalKB());
						addToPlanCost(sp, plans, newsel, cost);
					}

				} else {
					cost = subcost + est.getPlanCost(pjcand);
					addToPlanCost(sp, plans, pjcand, cost);
				}
				if (cost < mincost) {
					mincost = cost;
					bestcand = pjcand;
				}
			}
			// for a particular pjcand, only use its least cost plan as the best
			// plan
			// i.e. local best
			addToPlanCost(sp, plans, bestcand, mincost);
		}
		// i.e. global best
		bestplans.put(sp, bestcand);
	}

	private void bestForIntersection(JoinOP ip,
			Map<AlgebraicOP, Map<AlgebraicOP, Double>> plans,
			Map<AlgebraicOP, AlgebraicOP> bestplans)
			throws CostEstimateException {
		double mincost = Double.MAX_VALUE;
		AlgebraicOP bestcand = null;
		// Set<AlgebraicOP> res = applyRewritingSingle(ip);
		Set<AlgebraicOP> res = new HashSet<AlgebraicOP>();
		res.add(ip);
		for (AlgebraicOP pjcand : res) {
			double localcost = Double.MAX_VALUE;
			double cost = 0;
			AlgebraicOP ret = null;
			if (pjcand instanceof JoinOP) {
				// get the best ordering
				AlgebraicOP sub1 = pjcand.asIntersectionOP().getFirstSubQuery();
				AlgebraicOP sub2 = pjcand.asIntersectionOP()
						.getSecondSubQuery();
				if (!plans.containsKey(sub1)) {
					computePlans(sub1, plans, bestplans);
				}
				if (!plans.containsKey(sub2)) {
					computePlans(sub2, plans, bestplans);
				}
				for (AlgebraicOP s1 : plans.get(sub1).keySet()) {
					for (AlgebraicOP s2 : plans.get(sub2).keySet()) {
						JoinOP newip = new JoinOP(s1, s2);
						IntersectionGrouping ig = new IntersectionGrouping(
								newip, kb);
						ig.selectIntersectionOrder();
						AlgebraicOP posret = ig.getBestOrderPlan();
						cost = est.getPlanCost(posret)
								+ plans.get(sub1).get(s1)
								+ plans.get(sub2).get(s2);
						addToPlanCost(ip, plans, posret, cost);
						if (cost < localcost) {
							localcost = cost;
							ret = posret;
						}
					}
				}
				// add local min cost intersection plan
				// addToPlanCost(ip, plans, ret, localcost);
			} else if (pjcand instanceof SelectionOP) {
				// intersection may become selection, always add
				SelectionOP sp = pjcand.asSelectionOP();
				AlgebraicOP sub = sp.getSubquery();
				if (!plans.containsKey(sub)) {
					computePlans(sub, plans, bestplans);
				}
				ret = new SelectionOP(bestplans.get(sub),
						sp.getSelectionCondition(), kb);
				cost = est.getPlanCost(ret)
						+ plans.get(sub).get(bestplans.get(sub));
				addToPlanCost(ip, plans, ret, cost);
			} else {
				System.out.println("Uncaught plan for intersection: " + pjcand);
			}
			if (cost < mincost) {
				mincost = cost;
				bestcand = ret;
			}
		}
		// i.e. global best
		bestplans.put(ip, bestcand);
	}

	private void computePlans(AlgebraicOP alt,
			Map<AlgebraicOP, Map<AlgebraicOP, Double>> plans,
			Map<AlgebraicOP, AlgebraicOP> bestplans)
			throws CostEstimateException {
		if (plans.containsKey(alt)) {
			return;
		}
		if (alt instanceof ConstantOP) {
			addToPlanCost(alt, plans, alt, est.getPlanCost(alt));
			bestplans.put(alt, alt);
			return;
		} else if (alt instanceof PrimaryOP) {
			bestForPI(alt.asPrimaryOP(), plans, bestplans);
		} else if (alt instanceof ProjectionOP) {
			bestForProjection(alt.asProjectionOP(), plans, bestplans);
		} else if (alt instanceof SelectionOP) {
			bestForSelection(alt.asSelectionOP(), plans, bestplans);
		} else if (alt instanceof ScanOP) {
			bestForScan(alt.asScanOP(), plans, bestplans);
		} else if (alt instanceof JoinOP) {
			bestForIntersection(alt.asIntersectionOP(), plans, bestplans);
		}

	}

	// private double costOfIntersection(IntersectionOP ip,
	// Map<AlgebraicOP, Map<AlgebraicOP, Double>> plans, Map<AlgebraicOP,
	// AlgebraicOP> bestplans) {
	// double cost = 0;
	// AlgebraicOP sub1 = ip.getFirstSubQuery();
	// AlgebraicOP sub2 = ip.getSecondSubQuery();
	// if(sub1 instanceof IntersectionOP){
	// cost += costOfIntersection(sub1.asIntersectionOP(), plans, bestplans);
	// }else{
	// AlgebraicOP best = bestplans.get(sub1);
	// cost += plans.get(sub1);
	// }
	// if(sub2 instanceof IntersectionOP){
	// cost += costOfIntersection(sub2.asIntersectionOP(), plans, bestplans);
	// }else{
	// cost += costs.get(sub2);
	// }
	// return cost;
	// }

	private static enum SISelectionStrategy {
		ALL, POWERSET
	};

	// get a set of sets of SIs: powerset
	private Set<Set<SecondaryIndex>> selectSIByStrategy(
			SISelectionStrategy strat) {
		if (strat == null) {
			strat = SISelectionStrategy.POWERSET;
		}
		switch (strat) {
		case POWERSET:
			// Set<SecondaryIndex> sis =
			// SecondaryIndex.allCachedQueries.keySet();
			return LangPd.powerSet(this.sisForUse);
		default:
			break;
		}
		return null;
	}

	private Set<AlgebraicOP> applyRulesSaturate(Set<AlgebraicOP> op) {
		// System.out.println("Saturate: "+op);
		Set<AlgebraicOP> myop = new HashSet<AlgebraicOP>();
		myop.addAll(op);
		Set<AlgebraicOP> checked = new HashSet<AlgebraicOP>();
		while (myop.size() > 0) {
			AlgebraicOP mp = myop.iterator().next();
			myop.remove(mp);
			if (checked.contains(mp)) {
				continue;
			}
			checked.add(mp);
			Set<AlgebraicOP> ret = applyRewritingSingle(mp);
			if (!myop.containsAll(ret)) {
				ret.removeAll(checked);
				myop.addAll(ret);
			}
			myop.removeAll(checked);
		}
		return checked;
	}

	private Set<AlgebraicOP> applyRewritingSingle(AlgebraicOP sop) {
		Set<AlgebraicOP> ret = new HashSet<AlgebraicOP>();
		ret.add(sop);
		for (Set<SecondaryIndex> sis : this
				.selectSIByStrategy(SISelectionStrategy.POWERSET)) {
			for (RewriteRule rule : RuleService.getRulePlain(sop, sis, this)) {
				// System.err.println("apply "+rule.getClass());
				AlgebraicOP rop = rule.rewrite();
				if (rop != null) {
					ret.add(rop);
				}
			}
			if (!(sop instanceof PrimaryOP)) {
				// only PI needs multiple SIs
				break;
			}
		}
		return filterResults(sop, ret);
	}

	public Set<AlgebraicOP> filterResults(AlgebraicOP original,
			Set<AlgebraicOP> res) {
		if (original instanceof ProjectionOP) {
			return filterProjection(original.asProjectionOP(), res);
		} else if (original instanceof SelectionOP) {
			return filterSelection(original.asSelectionOP(), res);
		} else {
			return res;
		}
	}

	private Set<AlgebraicOP> filterSelection(SelectionOP sp,
			Set<AlgebraicOP> res) {
		Set<AlgebraicOP> ret = new HashSet<AlgebraicOP>();
		for (AlgebraicOP op : res) {
			if (op.equals(sp.getSubquery())) {
				// this selection can be removed
				ret.add(op);
			}
			// k-free op
			if (op instanceof SelectionOP) {
				if (op.getPlanEvalKB() == null
						|| !res.contains(op.getKFreeVersion())) {
					ret.add(op);
				}
			}
		}
		if (ret.size() < 1) {
			// System.out.println("not filtered, size:"+res.size());
			return res;
		}
		return ret;
	}

	private Set<AlgebraicOP> filterProjection(ProjectionOP pj,
			Set<AlgebraicOP> res) {
		Set<AlgebraicOP> ret = new HashSet<AlgebraicOP>();
		for (AlgebraicOP op : res) {
			if (!(op instanceof ProjectionOP)) {
				// potential this projection can be removed
				ret.add(op);
			}
			// add k-free ops as well
			if (op instanceof ProjectionOP) {
				if (op.getPlanEvalKB() == null
						|| !res.contains(op.getKFreeVersion())) {
					ret.add(op);
				}
				if (pj.getPlanEvalKB() != op.getPlanEvalKB()
						&& pj.getPrjDesc() != op.asProjectionOP().getPrjDesc()) {
					// two projection replaced by one Projection
					ret.add(op);
				}
			}
		}
		if (ret.size() < 1) {
			// System.out.println("not filtered, size:"+res.size());
			return res;
		}
		return ret;
	}

}
